package io.kiki.sba.registry.server.data_store.bootstrap;

import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.metaserver.ProvideData;
import io.kiki.sba.registry.common.model.slot.SlotsStatusResponse;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.metrics.ReporterUtils;
import io.kiki.sba.registry.metrics.TaskMetrics;
import io.kiki.sba.registry.net.NetUtil;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.lease.ClientInterfaceLeaseManager;
import io.kiki.sba.registry.server.data_store.providedata.FetchStopPushService;
import io.kiki.sba.registry.server.data_store.slot.SlotManager;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import io.kiki.sba.registry.server.shared.providedata.SystemPropertyProcessorManager;
import io.kiki.sba.registry.server.shared.remoting.AbstractServerHandler;
import org.glassfish.jersey.server.ResourceConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;

import javax.annotation.Resource;
import javax.ws.rs.Path;
import javax.ws.rs.ext.Provider;
import java.lang.annotation.Annotation;
import java.util.Collection;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;


@EnableConfigurationProperties
public class Bootstrap {
    private static final Logger logger = LoggerFactory.getLogger(Bootstrap.class);
    private final AtomicBoolean httpServerStarted = new AtomicBoolean(false);
    private final AtomicBoolean schedulerStarted = new AtomicBoolean(false);
    private final AtomicBoolean serverForSessionStarted = new AtomicBoolean(false);
    private final AtomicBoolean serverForDataSyncStarted = new AtomicBoolean(false);
    private final AtomicBoolean serverForMultiClusterDataSyncStarted = new AtomicBoolean(false);
    private final Retryer<Boolean> startupRetryer = RetryerBuilder.<Boolean>newBuilder().retryIfRuntimeException().retryIfResult(input -> !input).withWaitStrategy(WaitStrategies.exponentialWait(1000, 3000, TimeUnit.MILLISECONDS)).withStopStrategy(StopStrategies.stopAfterAttempt(10)).build();
    private final Retryer<Boolean> addBlacklistRetryer = RetryerBuilder.<Boolean>newBuilder().retryIfException().retryIfRuntimeException().withWaitStrategy(WaitStrategies.exponentialWait(1000, 3000, TimeUnit.MILLISECONDS)).withStopStrategy(StopStrategies.stopAfterAttempt(3)).build();
    @Autowired
    private ServerConfig serverConfig;
    @Autowired
    private MultiClusterDataServerConfig multiClusterDataServerConfig;
    @Autowired
    private MetaServerService metaServerService;
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private ResourceConfig jerseyResourceConfig;
    @Autowired
    private DataExchanger jerseyDataExchanger;
    @Autowired
    private DataExchanger boltDataExchanger;
    @Autowired
    private DataChangeEventCenter dataChangeEventCenter;
    @Autowired
    private ClientInterfaceLeaseManager clientInterfaceLeaseManager;
    @Resource(name = "serverHandlers")
    private Collection<AbstractServerHandler> serverHandlers;
    @Resource(name = "serverSyncHandlers")
    private Collection<AbstractServerHandler> serverSyncHandlers;
    @Resource(name = "remoteDataServerHandlers")
    private Collection<AbstractServerHandler> remoteDataServerHandlers;
    @Autowired
    private SystemPropertyProcessorManager systemPropertyProcessorManager;
    @Autowired
    private SlotManager slotManager;
    @Resource
    private FetchStopPushService fetchStopPushService;
    private Server server;
    private Server notifyServer;
    private Server dataSyncServer;
    private Server remoteDataSyncServer;
    private Server httpServer;


    public void start() {
        try {
            logger.info("begin start server");
            logger.info("release properties: {}", ServerEnv.getReleaseProps());
            logger.info("the configuration items are as follows: " + serverConfig.toString());

            ReporterUtils.enablePrometheusDefaultExports();

            openDataServer();

            openDataSyncServer();

            openMultiClusterDataSyncServer();

            openHttpServer();

            renewNode();
            fetchProviderData();

            systemPropertyProcessorManager.startFetchMetaSystemProperty();

            startScheduler();

            TaskMetrics.getInstance().registerBolt();

            postStart();
            Runtime.getRuntime().addShutdownHook(new Thread(this::doStop));

            logger.info("start server success");
        } catch (Throwable e) {
            throw new RuntimeException("start server error", e);
        }
    }

    private void postStart() throws Throwable {
        startupRetryer.call(() -> {
            logger.info("successful start data server, remove self from blacklist");
            metaServerService.removeSelfFromMetaBlacklist();
            return true;
        });
    }

    private void openDataServer() {
        try {
            if (serverForSessionStarted.compareAndSet(false, true)) {
                // open notify port first
                notifyServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getNotifyPort()), serverConfig.getLowWaterMark(), serverConfig.getHighWaterMark(), new ChannelHandler[0]);
                server = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getPort()), serverConfig.getLowWaterMark(), serverConfig.getHighWaterMark(), serverHandlers.toArray(new ChannelHandler[serverHandlers.size()]));
                dataChangeEventCenter.init();
                logger.info("Data server for session started! port:{}, notifyPort:{}", serverConfig.getPort(), serverConfig.getNotifyPort());
            }
        } catch (Exception e) {
            serverForSessionStarted.set(false);
            logger.error("Data server start error! port:{}", serverConfig.getPort(), e);
            throw new RuntimeException("Data server start error!", e);
        }
    }

    private void openDataSyncServer() {
        try {
            if (serverForDataSyncStarted.compareAndSet(false, true)) {
                dataSyncServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getSyncDataPort()), serverConfig.getLowWaterMark(), serverConfig.getHighWaterMark(), serverSyncHandlers.toArray(new ChannelHandler[serverSyncHandlers.size()]));
                logger.info("Data server for sync started! port:{}", serverConfig.getSyncDataPort());
            }
        } catch (Exception e) {
            serverForDataSyncStarted.set(false);
            logger.error("Data sync server start error! port:{}", serverConfig.getSyncDataPort(), e);
            throw new RuntimeException("Data sync server start error!", e);
        }
    }

    private void openMultiClusterDataSyncServer() {
        try {
            if (serverForMultiClusterDataSyncStarted.compareAndSet(false, true)) {
                remoteDataSyncServer = boltDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), multiClusterDataServerConfig.getSyncRemoteSlotLeaderPort()), multiClusterDataServerConfig.getSyncSlotLowWaterMark(), multiClusterDataServerConfig.getSyncSlotHighWaterMark(), remoteDataServerHandlers.toArray(new ChannelHandler[remoteDataServerHandlers.size()]));
                logger.info("Multi cluster data server for sync started! port:{}", multiClusterDataServerConfig.getSyncRemoteSlotLeaderPort());
            }
        } catch (Exception e) {
            serverForMultiClusterDataSyncStarted.set(false);
            logger.error("Multi cluster data sync server start error! port:{}", multiClusterDataServerConfig.getSyncRemoteSlotLeaderPort(), e);
            throw new RuntimeException("Multi cluster data sync server start error!", e);
        }
    }

    private void openHttpServer() {
        try {
            if (httpServerStarted.compareAndSet(false, true)) {
                bindResourceConfig();
                httpServer = jerseyDataExchanger.open(new Url(NetUtil.getLocalAddress().getHostAddress(), serverConfig.getHttpServerPort()), new ResourceConfig[]{jerseyResourceConfig});
                logger.info("Open http server port {} success!", serverConfig.getHttpServerPort());
            }
        } catch (Exception e) {
            httpServerStarted.set(false);
            logger.error("Open http server port {} error!", serverConfig.getHttpServerPort(), e);
            throw new RuntimeException("Open http server  error!", e);
        }
    }

    private void renewNode() {
        metaServerService.renewNode();
        // init session lease with first renew
        for (ClientInterfaceId clientInterfaceId : metaServerService.getClientInterfaceIdSet()) {
            clientInterfaceLeaseManager.renewal(clientInterfaceId);
        }
        metaServerService.startRenewer();
    }

    private void fetchProviderData() {
        ProvideData provideData = metaServerService.getProvideData(ValueConstants.DATA_SESSION_LEASE_SEC);
        Integer expireSec = ProvideData.toInteger(provideData);
        if (expireSec != null) {
            serverConfig.setSessionLeaseSecs(expireSec);
            logger.info("Fetch {}={}, update current config", ValueConstants.DATA_SESSION_LEASE_SEC, expireSec);
        }

        provideData = metaServerService.getProvideData(ValueConstants.DATA_DATUM_SYNC_SESSION_INTERVAL_SEC);
        Integer syncSessionIntervalSec = ProvideData.toInteger(provideData);
        if (syncSessionIntervalSec != null) {
            serverConfig.setSlotLeaderSyncSessionIntervalSecs(syncSessionIntervalSec);
            logger.info("Fetch {}={}, update current config", ValueConstants.DATA_DATUM_SYNC_SESSION_INTERVAL_SEC, syncSessionIntervalSec);
        }
    }

    private void startScheduler() {
        try {
            schedulerStarted.compareAndSet(false, true);
        } catch (Exception e) {
            schedulerStarted.set(false);
            logger.error("Data Scheduler start error!", e);
            throw new RuntimeException("Data Scheduler start error!", e);
        }
    }

    public void destroy() {
        doStop();
    }

    private void doStop() {
        try {
            logger.info("{} Shutting down Data Server..", new Date().toString());

            gracefulShutdown();

            stopHttpServer();
            stopServer();
            stopDataSyncServer();
            stopRemoteDataSyncServer();
            stopNotifyServer();
        } catch (Throwable e) {
            logger.error("Shutting down Data Server error!", e);
        }
        logger.info("{} Data server is now shutdown...", new Date().toString());
    }

    private void gracefulShutdown() {
        if (!serverConfig.isGracefulShutdown()) {
            logger.info("disable graceful shutdown, skip add blacklist");
            return;
        }

        try {
            addBlacklistRetryer.call(() -> {
                logger.info("[GracefulShutdown] add self to blacklist");
                metaServerService.addSelfToMetaBlacklist();
                return true;
            });
            addBlacklistRetryer.call(() -> {
                if (fetchStopPushService.isStopPushSwitch()) {
                    return true;
                }
                SlotsStatusResponse statusResponse = metaServerService.getSlotsStatus();
                if (statusResponse.isProtectionMode()) {
                    return true;
                }
                logger.info("[GracefulShutdown] wait no slot");
                if (slotManager.hasSlot()) {
                    throw new RuntimeException("current data server still own slot, waiting...");
                }
                return true;
            });
            logger.info("add data self to blacklist successfully");
        } catch (Throwable e) {
            logger.error("add blacklist failed:", e);
        }
    }

    private void stopHttpServer() {
        if (httpServer != null && httpServer.isOpen()) {
            httpServer.close();
        }
    }

    private void stopServer() {
        if (server != null && server.isOpen()) {
            server.close();
        }
    }

    private void stopDataSyncServer() {
        if (dataSyncServer != null && dataSyncServer.isOpen()) {
            dataSyncServer.close();
        }
    }

    private void stopRemoteDataSyncServer() {
        if (remoteDataSyncServer != null && remoteDataSyncServer.isOpen()) {
            remoteDataSyncServer.close();
        }
    }

    private void stopNotifyServer() {
        if (notifyServer != null && notifyServer.isOpen()) {
            notifyServer.close();
        }
    }

    private void bindResourceConfig() {
        registerInstances(Path.class);
        registerInstances(Provider.class);
    }

    private void registerInstances(Class<? extends Annotation> annotationType) {
        Map<String, Object> beans = applicationContext.getBeansWithAnnotation(annotationType);
        if (beans != null && !beans.isEmpty()) {
            beans.forEach((beanName, bean) -> jerseyResourceConfig.registerInstances(bean));
        }
    }

    public boolean getHttpServerStarted() {
        return httpServerStarted.get();
    }

    public boolean getSchedulerStarted() {
        return schedulerStarted.get();
    }

    public boolean getServerForSessionStarted() {
        return serverForSessionStarted.get();
    }

    public boolean getServerForDataSyncStarted() {
        return serverForDataSyncStarted.get();
    }


}
